//package com.yunhe.common.service.kafka;
//
//import com.fasterxml.jackson.databind.ObjectMapper;
//import com.yunhe.common.client.AuthorityClient;
//import com.yunhe.common.constant.AbnormalConsts;
//import com.yunhe.common.constant.KafkaConsts;
//import com.yunhe.common.domain.data.AbnormalEvent;
//import com.yunhe.common.domain.data.AbnormalEventRecord;
//import com.yunhe.common.domain.data.AbnormalEventRecordMeas;
//import com.yunhe.common.model.base.AbnormalStatusDTO;
//import com.yunhe.common.model.data.AbnormalRuleFormulaObject;
//import com.yunhe.common.model.data.Measurement;
//import com.yunhe.common.service.base.AbnormalStatusService;
//import com.yunhe.common.service.data.AbnormalEventRecordMeasService;
//import com.yunhe.common.service.data.AbnormalEventRecordService;
//import com.yunhe.common.service.data.AbnormalEventService;
//import com.yunhe.common.service.data.AlarmEventService;
//import com.yunhe.common.util.DateTimeUtils;
//import com.yunhe.common.util.RedisClient;
//import com.yunhe.common.util.StringUtil;
//import com.yunhe.common.util.CollectionUtil;
//import org.apache.kafka.clients.consumer.ConsumerRecord;
//import org.slf4j.Logger;
//import org.slf4j.LoggerFactory;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.kafka.annotation.KafkaListener;
//import org.springframework.stereotype.Component;
//import org.springframework.transaction.annotation.Transactional;
//
//import java.util.ArrayList;
//import java.util.HashMap;
//import java.util.List;
//import java.util.Map;
//
///**
// * @title: ClearAlarmListener
// * @Author Chen Lun
// * @Date: 2021/1/29
// */
//@Component
//public class ClearAlarmListener {
//    @Autowired
//    private AlarmEventService alarmEventService;
//    @Autowired
//    private AbnormalStatusService abnormalStatusService;
//    @Autowired
//    private AbnormalEventService abnormalEventService;
//    @Autowired
//    private AbnormalEventRecordService abnormalEventRecordService;
//    @Autowired
//    private AbnormalEventRecordMeasService abnormalEventRecordMeasService;
//    @Autowired
//    private AuthorityClient authorityClient;
//    @Autowired
//    private RedisClient redisClient;
//
//    private static final Logger logger = LoggerFactory.getLogger(ClearAlarmListener.class);
//
//    private static final ObjectMapper MAPPER = new ObjectMapper();
//
//    @Transactional
////    @KafkaListener(topics = KafkaConsts.Topic.ClearAlarm, groupId = KafkaConsts.Group.ClearAlarm)
//    public void clearAlarmByKafka(ConsumerRecord<?, String> record) {
//        try {
//            String key = record.value();
//            if (!StringUtil.isEmpty(key)) {
//                Map<String, Object> paramMap = redisClient.getObject(key, Map.class);
//                if (paramMap != null) {
//                    List<Object> objects = (List<Object>) paramMap.get("formula");
//                    Map<String, Number> valueMap = (Map<String, Number>) paramMap.get("value");
//                    Long rtime = (Long) paramMap.get("rtime");
//                    if (rtime == null) {
//                        return;
//                    }
//                    AbnormalStatusDTO eliminated = abnormalStatusService.findByName(AbnormalConsts.ABNORMAL_STATUS.ELIMINATED.getName());
//                    if (CollectionUtil.isNotEmpty(objects) && valueMap != null) {
//                        for (Object object : objects) {
//                            if (object == null) {
//                                continue;
//                            }
//                            AbnormalRuleFormulaObject formulaObject = MAPPER.convertValue(object, AbnormalRuleFormulaObject.class);
//                            List<Long> measPointNumbers = formulaObject.getMeasPointNumbers();
//                            //消除异常事件
//                            //查找相同告警规则和等级下未消除的事件
//                            List<AbnormalEvent> historyEvents =
//                                    abnormalEventService.findByAbnormalRuleIdAndAlarmLevelIdAndStatusId(formulaObject.getAbnormalRuleId(),
//                                            formulaObject.getAlarmLevelId(), eliminated.getId());
//                            if (CollectionUtil.isEmpty(historyEvents)) {
//                                continue;
//                            }
//                            Map<Long, Double> measPnValue = new HashMap<>();
//                            for (Long pn : measPointNumbers) {
//                                Number number = valueMap.get(String.valueOf(pn));
//                                Double pnValue = null;
//                                if (number != null) {
//                                    pnValue = number.doubleValue();
//                                }
//                                measPnValue.put(pn, pnValue);
//                            }
//                            if (measPnValue.containsValue(null)) {
//                                continue;
//                            }
//                            for (AbnormalEvent historyEvent : historyEvents) {
//                                String timeZone = authorityClient.getTimeZoneIdByStationId(historyEvent.getStationId()).getBody().getResults();
//                                historyEvent.setAbnormalStatusId(eliminated.getId());
//                                historyEvent.setAbnormalStatusTitle(eliminated.getTitle());
//                                historyEvent.setLatestProcessRtime(rtime);
//                                historyEvent.setLatestProcessTime(DateTimeUtils.getDateString(rtime, timeZone));
//                                abnormalEventService.update(historyEvent);
//                                createEventRecordMeas(historyEvent.getId(), measPnValue);
//                                //联动新增旧数据库
//                                try {
//                                    alarmEventService.updateEvent(historyEvent, eliminated);
//                                } catch (Exception e) {
//                                    logger.error("消除告警事件报错,{}", e.toString());
//                                }
//                            }
//                        }
//                    }
//                }
//            }
//        } catch (Exception e) {
//            logger.error("实时数据消除告警事件报错，{}，{}", record.value(), e.getStackTrace());
//        }
//    }
//
//    private void createEventRecordMeas(Long eventId, Map<Long, Double> pnValues) {
//        AbnormalEventRecord abnormalEventRecord = abnormalEventRecordService.findByAbnormalEventId(eventId);
//        if (abnormalEventRecord == null) {
//            logger.warn("未找到ID为{}的告警事件的记录", eventId);
//            return;
//        }
//        AbnormalEventRecordMeas recordMeas;
//        Measurement measurement;
//        List<AbnormalEventRecordMeas> recordMeasList = new ArrayList<>();
//        List<Measurement> measurementList = new ArrayList<>();
//        for (Map.Entry entry : pnValues.entrySet()) {
//            recordMeas = new AbnormalEventRecordMeas();
//            measurement = new Measurement();
//            recordMeas.setAbnormalEventRecord(abnormalEventRecord);
//            recordMeas.setPointNumber((Long) entry.getKey());
//            recordMeas.setValue((Double) entry.getValue());
//            recordMeasList.add(recordMeas);
//            measurement.setDtime(abnormalEventRecord.getDtime());
//            measurement.setPointNumber((Long) entry.getKey());
//            measurement.setVal((Double) entry.getValue());
//            measurementList.add(measurement);
//        }
//        if (CollectionUtil.isNotEmpty(recordMeasList)) {
//            abnormalEventRecordMeasService.create(recordMeasList);
//        }
//    }
//}
